package liveData.migration.toPhoenix.task.db;

import liveData.migration.toPhoenix.task.entity.FlatMessage;
import liveData.migration.toPhoenix.task.entity.VerifyResultDb;
import liveData.migration.toPhoenix.task.util.ClassHelper;
import liveData.migration.toPhoenix.task.util.ConstantUtil;
import liveData.migration.toPhoenix.task.util.DmlType;
import liveData.migration.toPhoenix.task.util.SqlHelper;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.schema.MetaDataClient;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.types.PDataType;
import org.apache.phoenix.util.SchemaUtil;

import java.lang.reflect.Field;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Map;

/**
 * @program: dataStreming
 * @description:
 * @author: sunteng
 * @create: 2020-02-27 09:37
 **/
@Slf4j
public class PhoenixJdbc extends RichSinkFunction<FlatMessage> {

    private Connection connection;
    private PreparedStatement upsertPreparedStatement;
    private PreparedStatement deletePreparedStatement;
    private VerifyResultDb verifyResultDb;
    private ClassHelper classHelper;
    private SqlHelper sqlHelper;
    private ObjectMapper mapper;
    private int length;
    private int[] columnTypes;
    private PTable ptable;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        connection = DriverManager.getConnection("jdbc:phoenix:10.10.12.215,10.10.12.216,10.10.12.129:2181:/hbase");
        connection.setAutoCommit(true);

        verifyResultDb = new VerifyResultDb();
        sqlHelper = new SqlHelper();
        classHelper = new ClassHelper();
        length = classHelper.getEntityProLength(verifyResultDb);
        ptable = getTableSchema(connection, ConstantUtil.TO_PHONEIX_table);
        columnTypes = getColumnSqlType();

        String upsertSql = "upsert into \"VERIFY_RESULT_PHOENIX\" values(" + classHelper.makeSql(verifyResultDb) + ")";
        upsertPreparedStatement = connection.prepareStatement(upsertSql);

        String deleteSql = "delete from \"VERIFY_RESULT_PHOENIX\" where \"id\" = ?";
        deletePreparedStatement = connection.prepareStatement(deleteSql);

        super.open(parameters);
        log.info("##open conn");
    }

    @Override
    public void close() throws Exception {

        if (upsertPreparedStatement != null) {
            upsertPreparedStatement.close();
        }

        if (deletePreparedStatement != null) {
            deletePreparedStatement.close();
        }

        if (connection != null) {
            connection.close();
        }

        log.info("##close conn");
        super.close();
    }

    @Override
    public void invoke(FlatMessage flatMessage, Context context) throws Exception {
        mapper = new ObjectMapper();
        Map<String, String> proMap = flatMessage.getData().get(0);

        log.info("## is DML data ? [{}]", flatMessage.getIsDdl());
        if (flatMessage.getType().equals(DmlType.DELETE.toString())) {
            log.info("##DELETE");
            deletePreparedStatement.setInt(1, mapper.convertValue(proMap, VerifyResultDb.class).getId());
            deletePreparedStatement.execute();

        } else if (flatMessage.getType().equals(DmlType.INSERT.toString()) || flatMessage.getType().equals(DmlType.UPDATE.toString())) {
            log.info("##UPSERT");

            String[] valueMap = new String[length];
            Field[] fields = classHelper.getEntityFields(verifyResultDb);
            for (int i = 0; i < fields.length; i++) {
                Field field = fields[i];
                field.setAccessible(true);
                valueMap[i] = proMap.get(field.getName());
            }

            sqlHelper.setupStatement(valueMap, length, columnTypes, upsertPreparedStatement);

            upsertPreparedStatement.executeUpdate();

        }
        log.info("##invoke");
    }

    private int[] getColumnSqlType() throws SQLException {
        Field[] fields = classHelper.getEntityFields(verifyResultDb);
        int[] types = new int[length];
        for (int i = 0; i < fields.length; i++) {
            Field field = fields[i];
            field.setAccessible(true);

            PDataType type = ptable.getColumnForColumnName(field.getName()).getDataType();
            types[i] = type.getSqlType();
            log.info("##Column name : " + field.getName() + ", sql type = " + type.getSqlType() + " " + type.getSqlTypeName());
        }
        return types;
    }

    private PTable getTableSchema(Connection conn, String fullTableName) throws SQLException {
        PhoenixConnection hconn = conn.unwrap(PhoenixConnection.class);
        MetaDataClient mdc = new MetaDataClient(hconn);
        String schemaName = SchemaUtil.getSchemaNameFromFullName(fullTableName);
        String tableName = SchemaUtil.getTableNameFromFullName(fullTableName);
        return mdc.updateCache(schemaName, tableName).getTable();
    }
}
